﻿using Confluent.Kafka;
using LJ.Abstractions.Config;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace LJ.EventBus.HornedSerpent.Transport.Kafka.Internal;

public class KafkaConnectionPool : IKafkaConnectionPool
{
    private readonly ConcurrentQueue<IProducer<string, byte[]>> _producerPool = [];


    private readonly EventBusTransportConfig _eventBusTransportConfig;
    private int _maxSize;
    private int _pCount;
    private bool disposedValue;

    public KafkaConnectionPool(IConfigProvider configProvider, ILogger<KafkaConnectionPool> logger)
    {
        _eventBusTransportConfig = configProvider.Get<EventBusTransportConfig>();

        _maxSize = _eventBusTransportConfig.ConnectionPoolSize;

        logger.LogDebug("Kafka servers: {servers}", _eventBusTransportConfig.Servers);
    }

    public IProducer<string, byte[]> RentProducer()
    {
        if (_producerPool.TryDequeue(out var producer))
        {
            Interlocked.Decrement(ref _pCount);

            return producer;
        }

        var config = new ProducerConfig(new Dictionary<string, string>(_eventBusTransportConfig.MainConfig))
        {
            BootstrapServers = _eventBusTransportConfig.Servers,
            QueueBufferingMaxMessages = 10,
            MessageTimeoutMs = 5000,
            RequestTimeoutMs = 3000
        };

        producer = new ProducerBuilder<string, byte[]>(config).Build();

        return producer;
    }

    public bool Return(IProducer<string, byte[]> producer)
    {
        if (Interlocked.Increment(ref _pCount) <= _maxSize)
        {
            _producerPool.Enqueue(producer);

            return true;
        }

        producer.Dispose();

        Interlocked.Decrement(ref _pCount);

        return false;
    }


    #region Dispose

    
    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                // TODO: 释放托管状态(托管对象)
                _maxSize = 0;

                while (_producerPool.TryDequeue(out var context))
                {
                    context.Dispose();
                }
            }

            // TODO: 释放未托管的资源(未托管的对象)并重写终结器
            // TODO: 将大型字段设置为 null
            disposedValue = true;
        }
    }

    // // TODO: 仅当“Dispose(bool disposing)”拥有用于释放未托管资源的代码时才替代终结器
    // ~KafkaConnectionPool()
    // {
    //     // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
    //     Dispose(disposing: false);
    // }

    public void Dispose()
    {
        // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }
    #endregion
}
